package com.bw.gmall.realjtime.dws.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.HbaseUtil;
import com.bw.gmall.realtime.common.util.RedisUtil;
import io.lettuce.core.RedisFuture;
import io.lettuce.core.api.StatefulRedisConnection;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.hadoop.hbase.client.AsyncConnection;

import java.io.IOException;
import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * @projectName: gmalll2206
 * @package: com.bw.gmall.realtime.dws.function
 * @className: AsyncDimFunction
 * @author: DuanYu
 * @description: TODO
 * @date: 2025/3/7 11:51
 * @version: 1.0
 */

public abstract class AsyncDimFunction<T> extends RichAsyncFunction<T,T> implements IsyncDim<T> {
    private AsyncConnection hBaseAsyncConnection;
    private StatefulRedisConnection<String, String> redisAsyncConnection;

    private String tableName;

    public AsyncDimFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        hBaseAsyncConnection = HbaseUtil.getHBaseAsyncConnection();
        redisAsyncConnection = RedisUtil.getRedisAsyncConnection();
    }

    @Override
    public void close() throws Exception {
        HbaseUtil.closeAsyncConnection(hBaseAsyncConnection);
        RedisUtil.closeRedisAsyncConnection(redisAsyncConnection);
    }

    @Override
    public void asyncInvoke(T t, ResultFuture<T> resultFuture) throws Exception {
        String dimId = getDimId(t);
        String redisKey = RedisUtil.getRedisKey(tableName, dimId);
        CompletableFuture.supplyAsync(new Supplier<JSONObject>() {
            @Override
            public JSONObject get() {
                String result = null;
                RedisFuture<String> stringRedisFuture = redisAsyncConnection.async().get(redisKey);
                try {
                    result = stringRedisFuture.get();
                } catch (InterruptedException  | ExecutionException e) {
                    resultFuture.completeExceptionally(e);
                    return null;
                }
                return JSON.parseObject(result);
            }
        }).thenApplyAsync(new Function<JSONObject, JSONObject>() {
            private JSONObject dimSkuInfo;
            @Override
            public JSONObject apply(JSONObject jsonObject) {
                if (jsonObject == null){
                    try {
                        dimSkuInfo = HbaseUtil.getAsyncCells(hBaseAsyncConnection, Constant.HBASE_NAMESPACE, tableName, dimId);
                        redisAsyncConnection.async().setex(redisKey,24*60*60,JSON.toJSONString(dimSkuInfo));
                    } catch (IOException e) {
                        resultFuture.completeExceptionally(e);
                        return null;
                    }
                }else {
                    dimSkuInfo = jsonObject;
                }
                return dimSkuInfo;
            }
        }).thenAccept(new Consumer<JSONObject>() {
            @Override
            public void accept(JSONObject jsonObject) {
                if (jsonObject != null){
                    getResult(t,jsonObject);
                }
                resultFuture.complete(Collections.singleton(t));
            }
        }).exceptionally(ex ->{
            resultFuture.completeExceptionally(ex);
            return null;
        });
    }
}
